package org.mq.mymq.nameserver.replication;

import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.NodeAckDto;
import org.mq.mymq.common.enums.NameServerEventCode;
import org.mq.mymq.nameserver.common.CommonCache;
import org.mq.mymq.nameserver.event.model.NodeReplicationMsgEvent;
import org.mq.mymq.nameserver.event.model.ReplicationMsgEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @ClassName NodeReplicationSendMsgTask
 * @Description 链式复制非尾节点发送数据任务
 * @Author jiarong_ye
 * @Date 2025/6/25 9:28
 * @Version 1.0
 */
public class NodeReplicationSendMsgTask extends ReplicationTask {

    private static final Logger log = LoggerFactory.getLogger(NodeReplicationSendMsgTask.class);

    public NodeReplicationSendMsgTask(String taskName) {
        super(taskName);
    }

    @Override
    void startTask() {
        while (true) {
            try {
                ReplicationMsgEvent replicationMsgEvent = CommonCache.getReplicationMsgQueueManager().getReplicationMsgQueue().take();
                Channel connectNodeChannel = CommonCache.getConnectNodeChannel();
                NodeReplicationMsgEvent nodeReplicationMsgEvent = new NodeReplicationMsgEvent();
                nodeReplicationMsgEvent.setMsgId(replicationMsgEvent.getMsgId());
                nodeReplicationMsgEvent.setServiceInstance(replicationMsgEvent.getServiceInstance());
                nodeReplicationMsgEvent.setType(replicationMsgEvent.getType());
                NodeAckDto nodeAckDto = new NodeAckDto();
                nodeAckDto.setBrokerCtx(replicationMsgEvent.getChannelHandlerContext());
                CommonCache.getNodeAckMap().put(replicationMsgEvent.getMsgId(), nodeAckDto);
                if (connectNodeChannel.isActive()) {
                    connectNodeChannel.writeAndFlush(new TcpMsg(NameServerEventCode.NODE_REPLICATION_MSG.getCode(),
                            JSON.toJSONBytes(nodeReplicationMsgEvent)));
                    log.info("链式复制-数据发送到下一个节点");
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}
